package pulsar_topic_reader

import (
	"context"
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"
	"go_19/initialize"
	"go_19/utils/array"
	"log"
	"strconv"
	"strings"
)

func TopicReader(topicReader initialize.TopicReader) {
	client := initialize.PulsarClient(topicReader.ClientOption)
	var mapMessageIdsArr []string

	if topicReader.MapMessageIds != "" {
		mapMessageIdsArr = strings.Split(topicReader.MapMessageIds, ",")
	}
	fmt.Println(mapMessageIdsArr)
	start := pulsar.EarliestMessageID()
	if topicReader.StartMessageID == "2" {
		start = pulsar.LatestMessageID()
	}
	if strings.Contains(topicReader.StartMessageID, ":") {
		startIdArr := strings.Split(topicReader.StartMessageID, ":")
		fmt.Println("startMessageId", startIdArr)
		ledgerID, _ := strconv.ParseInt(startIdArr[0], 10, 64)
		entryID, _ := strconv.ParseInt(startIdArr[1], 10, 64)
		batchIdx64 := int64(0)
		if len(startIdArr) >= 3 {
			batchIdx64Tmp, _ := strconv.ParseInt(startIdArr[2], 10, 64)
			batchIdx64 = batchIdx64Tmp
		}
		partitionIdx := int64(0)
		if len(startIdArr) > 3 {
			partitionIdxTmp, _ := strconv.ParseInt(startIdArr[3], 10, 64)
			partitionIdx = partitionIdxTmp
		}

		start = pulsar.NewMessageID(ledgerID, entryID, int32(batchIdx64), int32(partitionIdx))

	}
	fmt.Println(start)

	reader, err := client.CreateReader(pulsar.ReaderOptions{
		Topic:                  topicReader.Topic,
		Name:                   topicReader.ReaderName,
		StartMessageID:         start,
		SubscriptionRolePrefix: topicReader.SubscriptionRolePrefix,
		SubscriptionName:       topicReader.SubscriptionName,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer reader.Close()

	var msgMaped []string

	for reader.HasNext() {
		msg, err := reader.Next(context.Background())
		if err != nil {
			log.Fatal(err)
		}
		fmt.Println(msg.ID())
		if len(mapMessageIdsArr) > 0 {
			a := fmt.Sprintf("%v", msg.ID())
			for _, v := range mapMessageIdsArr {
				if a == v {
					fmt.Printf("-- msgId: %v -- publishTime: %s -- content: %s\n",
						msg.ID(), msg.PublishTime().Format("2006-01-02 15:04:05.999999999"), string(msg.Payload()))
					msgMaped = append(msgMaped, a)
					break
				}
			}
			msgMaped = array.UniqueString(append(msgMaped, a))
			if len(msgMaped) == len(mapMessageIdsArr) {
				fmt.Println()
				fmt.Println("-- 已查找到所有信息", mapMessageIdsArr)
				break
			}
		} else {
			fmt.Printf("-- msgId: %v -- publishTime: %s -- content: %s\n",
				msg.ID(), msg.PublishTime().Format("2006-01-02 15:04:05.999999999"), string(msg.Payload()))
		}

	}

}
